package org.ykhl.log.commons.sender;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.ykhl.mq.commons.Constants;
import org.ykhl.mq.commons.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;


/**
 * Created by zkf on 2015/12/10.
 */
public class LogSenderKafkaImpl extends LogSender {
    private static final Logger LOG = LoggerFactory.getLogger(LogSenderKafkaImpl.class);
    private Producer<String, String> producer;

    @Override
    public void init() {
        LOG.info("LogSenderKafkaImpl start to init...");
        Properties props = Utils.genKafkaProducerProps(mqNameServerAddress, useSsl);
        //尽量一次多发几条消息，最小发送大小400 bytes
//        props.put(ProducerConfig.BATCH_SIZE_CONFIG, Constants.KAFKA_BATCH_SIZE_IN_BYTES);
        props.put(ProducerConfig.ACKS_CONFIG, "0");
        producer = new KafkaProducer<>(props);

        LOG.info("init - done");
    }

    @Override
    public void destroy() {
        producer.close();

        LOG.info("destroy - done");
    }

    @Override
    public void flush() {
        producer.flush();
    }

    @Override
    protected void sendMessage(String jsonLogMessage) throws Exception {
        ProducerRecord<String, String> record = new ProducerRecord<>(Constants.MQ_LOG_TOPIC, jsonLogMessage);
        producer.send(record);
    }
}
